package main

import (
	"sync"
	"time"
)

type(
	subscribe chan interface{}
	topicFunc func(v interface{})bool
)

type Publisher struct {
	m sync.RWMutex
	buffer int
	timeout time.Duration
	subscribe map[subscribe]topicFunc
}

func NewPublisher(timeout time.Duration, buffer int) *Publisher {
	return &Publisher{
		buffer:buffer,
		timeout: timeout,
		subscribe: make(map[subscribe]topicFunc),
	}
}

func (p *Publisher) publish(v interface{}) {
	p.m.RLock()
	defer p.m.RUnlock()
	var wg sync.WaitGroup
	for sub, topic := range p.subscribe {
		wg.Add(1)
		go p.sendTopic(sub, topic, v, &wg)
	}
	wg.Wait()
}

func (p *Publisher) sendTopic(sub subscribe, topic topicFunc, v interface{}, wg *sync.WaitGroup){
	defer wg.Done()
	//filter
	if topic !=nil && !topic(v){
		return
	}

	select {
	case sub <- v:
	case <-time.After(p.timeout):
	}
}

//添加一个订阅者
func (p *Publisher)SubscribeTopic(topic topicFunc) chan interface{}{
	p.m.Lock()
	defer p.m.Unlock()

	ch := make(chan interface{}, 10)
	p.subscribe[ch] = topic
	return ch
}


